package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.DimJoinFunction;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @author Felix
 * @date 2024/11/09
 * 使用旁路缓存进行维度关联
 */
public abstract class DimMapFunction<T> extends RichMapFunction<T, T> implements DimJoinFunction<T> {
    private Connection hBaseConn;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hBaseConn);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public T map(T obj) throws Exception {
        //根据流中对象获取要关联的维度主键
        String rowkey = getRowKey(obj);
        //先从Redis中获取维度数据
        JSONObject dimJsonObj = RedisUtil.readDim(jedis, getTableName(), rowkey);
        if (dimJsonObj != null) {
            //如果从Redis中获取到了维度数据，直接将其作为返回值进行返回(缓存命中)
            System.out.println("~~~从Redis中获取到表" + getTableName() + "的" + rowkey + "数据~~~");
        } else {
            //如果从Redis中没有获取到要关联的维度，发送请求到HBase中进行查询
            dimJsonObj = HBaseUtil.getRow(hBaseConn, Constant.HBASE_NAMESPACE,getTableName(),rowkey, JSONObject.class);
            if(dimJsonObj != null){
                //将查询到的维度数据放到Redis中缓存起来
                System.out.println("~~~从HBase中获取到表" + getTableName() + "的" + rowkey + "数据~~~");
                RedisUtil.writeDim(jedis,getTableName(),rowkey,dimJsonObj);
            }else{
                System.out.println("~~~没有找到表" + getTableName() + "的" + rowkey + "数据~~~");
            }
        }
        //将维度属性补充到流中对象上
        if(dimJsonObj != null){
            addDims(obj,dimJsonObj);
        }
        return obj;
    }

}
